package net.bwie.jtp.dim.function;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashSet;
import java.util.Set;

/*
 * DimDataFilterFunction类用于过滤维度数据
 * 继承RichFilterFunction<String>实现字符串类型的数据过滤
 * 
 * 参数:
 * dimSet - 需要过滤的维度表集合
 * 
 * 返回值:
 * boolean - 过滤结果，true保留数据，false过滤掉数据
 */
public class DimDataFilterFunction extends RichFilterFunction<String> {
    // 存储需要过滤的维度表名称集合
    private Set<String> dimSet;

    // 允许的操作类型集合，包含insert/update/delete三种操作类型
    private final Set<String> operateTypeSet = new HashSet<String>(){
        {
            add("insert");add("update");add("delete");
        }
    };

    /*
     * 构造函数
     * 
     * 参数:
     * dimSet - 需要过滤的维度表集合
     */
    public DimDataFilterFunction(Set<String>dimSet){
        this.dimSet = dimSet;
    }

    /*
     * filter方法实现具体的过滤逻辑
     * 通过解析JSON字符串，获取表名和操作类型，判断是否在允许的维度表和操作类型集合中
     * 
     * 参数:
     * value - JSON格式的字符串数据
     * 
     * 返回值:
     * boolean - 过滤结果，true保留数据，false过滤掉数据
     */
    @Override
    public boolean filter(String value) throws Exception {
        // 解析JSON字符串
        JSONObject jsonObject = JSON.parseObject(value);
        // 获取表名
        String table_name = jsonObject.getString("table_name");
        // 获取操作类型
        String operateType = jsonObject.getString("operate_type");
        // 判断表名是否在允许的维度表集合中，且操作类型是否在允许的操作类型集合中
        return dimSet.contains(table_name) && operateTypeSet.contains(operateType);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }























}
